package local.blueheart.springcloud.rabbitmq.route;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import local.blueheart.springcloud.rabbitmq.common.BHCustomConsumer;
import local.blueheart.springcloud.rabbitmq.util.MessageBrokerHelper;
import org.apache.commons.lang3.RandomUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public class RouteQueueHelper {

    private static final Logger LOGGER = LoggerFactory.getLogger(RouteQueueHelper.class);

    private static final String EXCHANGE_NAME = "test_route";

    private static final String QUEUE_C1_NAME = "test_route_queue1";

    private static final String QUEUE_C2_NAME = "test_route_queue2";

    private static final String[] severitys = {"error","info","warning"};

    private MessageBrokerHelper messageBrokerHelper;


    public RouteQueueHelper() {
        String host = "127.0.0.1";
        int port = 5672;
        String vHost = "/bhmmr";
        String username = "bh_mmr";
        String password = "sunny6150";
        messageBrokerHelper = new MessageBrokerHelper(host,port,vHost,username,password);
    }

    private String getRandomSeverity(){
        int r = RandomUtils.nextInt(0,severitys.length);
        return severitys[r];
    }

    public void sender(String body, int countTime){
        Connection connection = messageBrokerHelper.getConnection();
        Channel channel = messageBrokerHelper.getChannel(connection,EXCHANGE_NAME,"direct");
        try {
            int senderTime = countTime ;
            for (int i=0;i<senderTime;i++) {
                String severity = getRandomSeverity();
                String msg =  body + i + " ["+severity+"]";
                channel.basicPublish(EXCHANGE_NAME,severity,null,msg.getBytes());
                Thread.sleep(20);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            messageBrokerHelper.closeChannel(channel);
            messageBrokerHelper.closeConnection(connection);
        }

    }


    public void startConsumer1(final String... severitys) {
        new Thread(new Runnable() {
            public void run() {
                LOGGER.debug("Consumer1 starting.....["+severitys+"]");
                Connection connection = messageBrokerHelper.getConnection();
                Channel channel = messageBrokerHelper.getChannel(connection, QUEUE_C1_NAME);
                try {
                    for(String severity:severitys) {
                        channel.queueBind(QUEUE_C1_NAME, EXCHANGE_NAME, severity);
                    }
                    Consumer consumer = new BHCustomConsumer(channel, "C1");
                    channel.basicConsume(QUEUE_C1_NAME, true, consumer);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public void startConsumer2(final String... severitys){
        new Thread(new Runnable() {
            public void run() {
                LOGGER.debug("Consumer2 starting.....["+severitys+"]");
                Connection connection = messageBrokerHelper.getConnection();
                Channel channel = messageBrokerHelper.getChannel(connection, QUEUE_C2_NAME);
                try {
                    for(String severity:severitys) {
                        channel.queueBind(QUEUE_C2_NAME, EXCHANGE_NAME, severity);
                    }
                    Consumer consumer = new BHCustomConsumer(channel, "C2");
                    channel.basicConsume(QUEUE_C2_NAME, true, consumer);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public static void main(String[] args){
        RouteQueueHelper exchangeQueueHelper = new RouteQueueHelper();
        exchangeQueueHelper.startConsumer1("error");
        exchangeQueueHelper.startConsumer2("info","error","warning");
        exchangeQueueHelper.sender("Hello Exchange Rabbit MQ",1000);
    }



}
